<?php


namespace App\Http\Library;

use App\Config\QueueConfig;

class Queue
{
    /*重试上限*/
    private static $redo = 10;
    /*任务队列阻塞等待时间(秒),这里阻塞时间建议不要超过60s,设置太长时间会影响延迟队列任务执行*/
    private static $wait = 3;
    /*延时任务下次扫描时间(不需要改,默认即可)*/
    private static $clockTime = 0;
    /*注册任务队列*/
    private static $register = QueueConfig::REGISTER;
    public static $traceId = '-';
    public static $microtime = null;
    /*任务队列空闲回调*/
    private static $free = QueueConfig::FREE;
    /*空闲回调任务间隔时间记录*/
    private static $freeKeepTime = [];

    /**
     * 即时任务-队列头部插入
     *
     * @param array  $data
     * @param string $_task
     *
     * @return mixed
     * @throws \Exception
     */
    public static function lpush(array $data, string $_task = '')
    {
        if (!empty($_task)) {
            $data['_task'] = $_task;
        }
        if (empty($data['_task'])) {
            throw new \Exception('缺失任务_task');
        }
        if (empty($data['_trace_id'])) {
            $data['_trace_id'] = request()->header('X-Trace-Id', '-');
        }
        return Cache::lpush(QueueConfig::queue(), $data);
    }


    /**
     * 即时任务-队列尾部追加-批量
     *
     * @param array  $data
     * @param string $_task
     * @param array  $param
     *
     * @throws \Exception
     */
    public static function rpushList(array $data, string $_task, array $param = [])
    {
        if (empty($_task)) {
            throw new \Exception('缺失任务_task');
        }
        $traceId = request()->header('X-Trace-Id', '-');

        foreach ($data as $v) {
            if (is_object($v)) {
                $v = (array) $v;
            }
            /*如果有自主定义的参数,则合并*/
            if (!empty($param)) {
                $v = array_merge($v, $param);
            }
            Cache::rpush(QueueConfig::queue(), array_merge($v, ['_task' => $_task, '_trace_id' => $traceId]));
        }
    }


    /**
     * 即时任务-队列尾部追加
     *
     * @param array  $data
     * @param string $_task
     *
     * @return mixed
     * @throws \Exception
     */
    public static function rpush(array $data, string $_task = '')
    {
        if (!empty($_task)) {
            $data['_task'] = $_task;
        }
        if (empty($data['_task'])) {
            throw new \Exception('缺失任务_task');
        }
        if (empty($data['_trace_id'])) {
            $data['_trace_id'] = request()->header('X-Trace-Id', '-');
        }
        return Cache::rpush(QueueConfig::queue(), $data);
    }


    /**
     * 延时任务
     *
     * @param array  $data  数据
     * @param int    $defer 延迟执行时间(秒)
     * @param string $_task 任务task
     *
     * @return mixed
     */
    public static function clock(array $data, int $defer = 60, string $_task = '')
    {
        /*延时任务会生成当前任务累计执行次数*/
        $data['_redo'] = empty($data['_redo']) ? 1 : ($data['_redo'] + 1);
        if ($data['_redo'] > self::$redo) {
            throw new \Exception("队列重试上限：_redo={$data['_redo']}");
        }
        if (!empty($_task)) {
            $data['_task'] = $_task;
        }
        if (empty($data['_task'])) {
            throw new \Exception('缺失任务_task');
        }
        if (empty($data['_trace_id'])) {
            $data['_trace_id'] = request()->header('X-Trace-Id', '-');
        }
        return Cache::zadd(QueueConfig::clock(), time() + $defer, $data);
    }


    /**
     * 获取当前任务_traceId
     *
     * @param array $data
     *
     * @return string
     */
    private static function _traceId(array $data = []) :string
    {
        if (!empty($data['_trace_id']) && $data['_trace_id'] !== '-') {
            return $data['_trace_id'];
        }
        /*如果_traceId不存在,则生成一个*/
        return 'job' . date('YmdHis') . 'oo' . uniqid();
    }


    /**
     * 执行任务队列
     */
    public function handle()
    {
        /*设置任务队列执行时长*/
        $exTime = time() + config('ext.queue_restart_ttl', 10800);
        while ($exTime > time()) {
            try {
                /*获取一条待执行的任务*/
                $data = self::_init();
                if (self::_continue($data, $_task) === true) {
                    continue;
                }
                /*记录当前任务_trace_id*/
                self::$traceId = self::_traceId($data);
                /*记录当前任务开始时间*/
                self::$microtime = microtime(true);
                /*调用已注册任务*/
                if (isset(self::$register[$_task])) {
                    call_user_func([self::$register[$_task], 'main'], $data);
                }
                self::$traceId = '-';
                self::$microtime = null;
            } catch (\Exception $exception) {
                $err = [
                    '[队列任务异常]',
                    'line' => $exception->getLine(),
                    'file' => $exception->getFile(),
                    'msg'  => $exception->getMessage()
                ];
                log_arr(array_merge($err, [
                    'data' => serialize($data)
                ]), 'queue');
                if (config('app.debug') === true) {
                    var_dump($err);
                } else {
                    /*调用预警推送通知*/
                    Paas::yach($err);
                }
                throw new \Exception($exception->getMessage(), $exception->getCode());
            }
        }
    }


    /**
     * 获取一条待执行的任务
     *
     * @return mixed
     */
    private static function _init()
    {
        /*定时任务*/
        self::_clock();
        /*从任务队列取一条待处理的数据*/
        return Cache::blpop(QueueConfig::queue(), self::$wait);
    }


    /**
     * 延迟任务处理
     */
    private static function _clock()
    {
        $time = time();
        /*是否需要触发延迟任务检查*/
        if (self::$clockTime > $time) {
            return;
        }
        /*设置下次延迟任务检查时间*/
        self::$clockTime = $time + 5;
        $list = Cache::zrangebyscore(QueueConfig::clock(), 0, $time);
        if (empty($list)) {
            return;
        }
        try {
            /*监视延迟队列是数据否发生变更*/
            Cache::watch([QueueConfig::clock()]);
            /*事务处理数据*/
            Cache::transaction(function() use ($list) {
                foreach ($list as $val) {
                    /*从集合清除当前数据*/
                    Cache::zrem(QueueConfig::clock(), $val);
                    /*写入任务队列*/
                    Cache::rpush(QueueConfig::queue(), $val);
                }
            });
        } catch (\Exception $exception) {
            /*释放监视*/
            Cache::unwatch([QueueConfig::clock()]);
            throw new \Exception($exception->getMessage(), $exception->getCode());
        }
    }


    /**
     * 是否存在需要处理的数据
     *
     * @param $data
     *
     * @return bool
     */
    private static function _continue($data, &$_task)
    {
        if (!empty($data['_task'])) {
            $_task = $data['_task'];
            return false;
        }
        /*sleep(0.1)*/
        usleep(100000);
        /*任务列队空闲回调函数*/
        self::free();
        return true;
    }


    /**
     * 任务列队空闲回调函数
     *
     * @return int
     */
    private static function free()
    {
        if (empty(self::$free)) {
            return 0;
        }
        foreach (self::$free as $key => $val) {
            /*初始化执行时间*/
            if (empty(self::$freeKeepTime[$key])) {
                self::$freeKeepTime[$key] = time();
            }
            /*如果不满足执行时间,则直接返回*/
            if (self::$freeKeepTime[$key] > time()) {
                continue;
            }
            /*间隔时间*/
            $keepTime = (int) ($val['keep_time'] < 10 ? 10 : $val['keep_time']);
            /*计算下次触发时间*/
            self::$freeKeepTime[$key] = time() + $keepTime;
            /*
             * 如果没有拿到锁,则跳过处理
             * 设置死锁自动释放时间为 $keepTime 秒
             * */
            if (Lock::lock($key, 0, $keepTime) === false) {
                continue;
            }
            /*执行回调*/
            call_user_func([$val['func'], 'call']);
        }
        return 0;
    }

}





/**
 * Queue.php
 *
 * 说明:
 *
 * 修改历史
 * ----------------------------------------
 * 2021/1/8 16:48:00   操作:创建
 **/
